package com.atguigu.day09;

import com.atguigu.utils.UserBehavior;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.shaded.guava18.com.google.common.base.Charsets;
import org.apache.flink.shaded.guava18.com.google.common.hash.BloomFilter;
import org.apache.flink.shaded.guava18.com.google.common.hash.Funnels;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;
import java.util.HashSet;

import static org.apache.flink.table.api.Expressions.$;

// 布隆过滤器
// 每小时的uv统计
// uv是pv按照userId进行去重后的结果
public class Example7 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<UserBehavior> stream = env
                .readTextFile("/home/zuoyuan/flinktutorial0722/src/main/resources/UserBehavior.csv")
                .map(new MapFunction<String, UserBehavior>() {
                    @Override
                    public UserBehavior map(String value) throws Exception {
                        String[] arr = value.split(",");
                        return new UserBehavior(
                                arr[0], arr[1], arr[2], arr[3],
                                Long.parseLong(arr[4]) * 1000L
                        );
                    }
                })
                .filter(r -> r.type.equals("pv"))
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<UserBehavior>forMonotonousTimestamps()
                                .withTimestampAssigner(new SerializableTimestampAssigner<UserBehavior>() {
                                    @Override
                                    public long extractTimestamp(UserBehavior element, long recordTimestamp) {
                                        return element.ts;
                                    }
                                })
                );

        // 获取表执行环境
        EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().build();
        StreamTableEnvironment streamTableEnvironment = StreamTableEnvironment.create(env, settings);

        // 将数据流转换成动态表
        Table table = streamTableEnvironment
                .fromDataStream(
                        stream,
                        $("userId"),
                        $("itemId"),
                        $("categoryId").as("cid"),
                        $("type"),
                        // rowtime表示ts是事件时间
                        $("ts").rowtime()
                );

        // 将动态表转换回数据流
        // DataStream<Row> rowDataStream = streamTableEnvironment.toDataStream(table);
        // rowDataStream.print();
        // +I[163756, 4146871, 1029459, pv, 2017-11-26 03:20:37.0]
        // +I表示Insert

        // 在动态表上进行查询
        // 计算的是item view count per window
        // 将动态表注册为临时视图
        streamTableEnvironment.createTemporaryView("userbehavior", table);
        // HOP表示滑动窗口，第一个参数是时间戳的字段名，第二个参数是滑动距离，第三个参数是窗口长度
        // COUNT是全窗口聚合
//        Table result = streamTableEnvironment
//                .sqlQuery(
//                        "SELECT itemId, COUNT(itemId) as cnt," +
//                                " HOP_START(ts, INTERVAL '5' MINUTES, INTERVAL '1' HOURS) as windowStartTime, " +
//                                " HOP_END(ts, INTERVAL '5' MINUTES, INTERVAL '1' HOURS) as windowEndTime " +
//                                "FROM userbehavior GROUP BY itemId, HOP(ts, INTERVAL '5' MINUTES, INTERVAL '1' HOURS)"
//                );

//        streamTableEnvironment.toChangelogStream(result).print();

        String innerSQL = "SELECT itemId, COUNT(itemId) as cnt," +
                " HOP_START(ts, INTERVAL '5' MINUTES, INTERVAL '1' HOURS) as windowStartTime, " +
                " HOP_END(ts, INTERVAL '5' MINUTES, INTERVAL '1' HOURS) as windowEndTime " +
                "FROM userbehavior GROUP BY itemId, HOP(ts, INTERVAL '5' MINUTES, INTERVAL '1' HOURS)";

        // 按照窗口结束时间分区，然后按照浏览量降序排列
        String midSQL = "SELECT *, ROW_NUMBER() OVER (PARTITION BY windowEndTime ORDER BY cnt DESC) as row_num" +
        " FROM (" + innerSQL + ")";

        String outerSQL = "SELECT * FROM (" + midSQL + ") WHERE row_num <= 3";

        Table result = streamTableEnvironment.sqlQuery(outerSQL);

        // 当sql查询中存在聚合操作时，必须使用toChangelogStream
        streamTableEnvironment.toChangelogStream(result).print();

        env.execute();
    }
}
